package com.bd08.flink.demo.jing

import org.apache.flink.api.scala.{ExecutionEnvironment, createTypeInformation}
import org.apache.flink.core.fs.FileSystem

object Customers extends App {

  case class Customer(id:String,name:String,address:String)
  val benv = ExecutionEnvironment.getExecutionEnvironment
  //读取CSV文件中的first_name,last_name,city字段
  val csvInput = benv.readCsvFile[Customer]("D:\\yahu\\bugeikan\\src\\main\\resources\\customers.csv", includedFields = Array(1,2,6),quoteCharacter = '"')
  csvInput
    .filter(customer => customer.address.contains("Chicago"))

    .writeAsCsv("D:\\yahu\\bugeikan\\src\\main\\resources\\Chicago","\n",",",FileSystem.WriteMode.OVERWRITE)

  csvInput.groupBy("name")
    .reduceGroup(group=>{
      (group.next().name,group.size)
    })
    .writeAsCsv("D:\\yahu\\bugeikan\\src\\main\\resources\\tongming","\n",",",FileSystem.WriteMode.OVERWRITE)


  csvInput.groupBy("name","address")
    .reduceGroup(group=>{
      val e = group.next()
      (e.name+"|"+e.address,group.size)
    })
    .writeAsCsv("D:\\yahu\\bugeikan\\src\\main\\resources\\tongming2","\n",",",FileSystem.WriteMode.OVERWRITE)


  benv.execute

}
